package com.pie4cloud.pie.job.executor.utils;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.util.Arrays;
import java.util.Comparator;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicStampedReference;

import com.pie4cloud.pie.job.api.entity.ConfigFtp;
import lombok.Getter;
import lombok.Setter;
import org.apache.commons.net.ftp.FTP;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPFile;
import org.apache.commons.net.ftp.FTPReply;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;


/**
 * FTP上传下载
 *
 * @author ShenHuaJie
 * @version 2016年5月20日 下午3:19:19
 */
public class FtpClient {
	private Logger logger = LogManager.getLogger();
	private static final byte[] LOCK = { 0 };
	private static final byte[] FILELOCK = { 1 };
	@Getter
	@Setter
	private  FTPClient ftpClient = null;
	private static  ConcurrentHashMap<Long, FtpClient> ftps=new ConcurrentHashMap<>();
	private static  ConcurrentHashMap<Long, Long> ftpsTimeOut=new ConcurrentHashMap<>();
	@Getter
	@Setter
	private ConfigFtp configFtp=null;
	private AtomicStampedReference<Integer> atomic = new AtomicStampedReference<Integer>(0,0);


	private static ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(1);


	public FtpClient getFtpClient(ConfigFtp configFtp){
		FtpClient ftpClient=null;
		if(null!=ftps.get(configFtp.getFtpId())&&ftps.get(configFtp.getFtpId()).ftpClient.isConnected()){
			ftpClient=ftps.get(configFtp.getFtpId());
		}else {
			synchronized (LOCK) {
				if(null!=ftps.get(configFtp.getFtpId())&&ftps.get(configFtp.getFtpId()).ftpClient.isConnected()){
					ftpClient=ftps.get(configFtp.getFtpId());
				}else {
					if(null!=ftps.get(configFtp.getFtpId())){
						try {
							ftps.get(configFtp.getFtpId()).ftpClient.logout();
						} catch (IOException e) {
							e.printStackTrace();
						}
						ftps.remove(configFtp.getFtpId());
					}
					ftpClient = new FtpClient(configFtp.getFtpAddress(), configFtp.getFtpPort().intValue(), configFtp.getFtpUserName(), configFtp.getFtpPwd());
					ftps.put(configFtp.getFtpId(), ftpClient);
					ftpsTimeOut.put(configFtp.getFtpId(),System.currentTimeMillis());
					scheduledThreadPool.schedule(()->{
                           this.ftpTimeOut();
					},30, TimeUnit.MINUTES);
				}
			}
		}
		ftpClient.setConfigFtp(configFtp);
		return ftpClient;
	}

	public void ftpTimeOut(){
		long now =System.currentTimeMillis();
		long old =ftpsTimeOut.get(configFtp.getFtpId());
		if((now-old)>=30*1000*60&&atomic.getReference()==0){
			if(null!=ftps.get(configFtp.getFtpId())){
				ftps.get(configFtp.getFtpId()).close();
				ftps.remove(configFtp.getFtpId());
			}
		}else {
			scheduledThreadPool.schedule(()->{
				this.ftpTimeOut();
			},30, TimeUnit.MINUTES);
		}
	}

	public static void main(String[] args) throws Exception {
		String host = "10.110.173.233";
		int port = 21;
		String username = "ftpuser";
		String password = "ftpuser";
		String localUpPath = "D://222.jpg";
		String localDnPath = "D:/";
		String remotePath = "/CMADAAS/DATA/SURF/MERGE/CMPA/FAST/HOR/2021/20210507";
		FtpClient ftpClient = new FtpClient(host, port, username, password);

		// FTP上传文件
		//ftpClient.uploadFiles(new File(localUpPath), localUpPath,"2");
		// FTP下载文件
		ftpClient.downLoadFile(remotePath, localDnPath);

		ftpClient.close();
	}

	public FtpClient() {
	}

	/**
	 * 初始化
	 *
	 * @param host IP
	 * @param port 端口
	 * @param username 用户名
	 * @param password 密码
	 * @throws FtpException e
	 */
	public FtpClient(String host, int port, String username, String password) throws FtpException {
		init(host, port, username, password);
	}


	/**
	 * 获取FTP连接
	 *
	 * @param
	 * @param
	 * @param password
	 * @throws FtpException e
	 */
	private void init(String host, int port, String username, String password) throws FtpException {
		synchronized (LOCK) {
			if (ftpClient == null) {
				ftpClient = new FTPClient();
			}
			try {
				ftpClient.connect(host, port);// 连接FTP服务器
			} catch (Exception e) {
				throw new FtpException("FTP[" + host + ":" + port + "]连接失败!", e);
			}
			// 如果采用默认端口，可以使用ftp.connect(url)的方式直接连接FTP服务器
			if (FTPReply.isPositiveCompletion(ftpClient.getReplyCode())) {
				try {
					ftpClient.login(username, password);
				} catch (Exception e) {
					throw new FtpException("FTP用户[" + username + "]登陆失败!", e);
				}
			} else {
				throw new FtpException("FTP连接出错!");
			}
			logger.info("用户[" + username + "]登陆[" + host + "]成功.");
			try {
				// 设置被动模式
				ftpClient.enterLocalPassiveMode();
				ftpClient.setFileTransferMode(FTP.STREAM_TRANSFER_MODE);
				ftpClient.setFileType(FTP.BINARY_FILE_TYPE);
			} catch (Exception e) {
				logger.error("", e);
				throw new FtpException("FTP初始化出错!", e);
			}
		}
	}

	/**
	 * 关闭FTP客户端
	 * @throws FtpException e
	 */
	public void close() throws FtpException {
		synchronized (LOCK) {
			try {
				ftpClient.logout();
			} catch (IOException e) {
				logger.error("", e);
				ftpClient = null;
				throw new FtpException("FTP退出登录出错!", e);
			}
		}
	}

	/**
	 * 上传
	 *
	 * @param remotePath 上传目录
	 * @param localPath 本地目录
	 * @return boolean
	 * @throws FtpException e
	 */
	public boolean uploadFile(String remotePath, String localPath,String fileName) throws FtpException {
		atomic.compareAndSet(atomic.getReference(),atomic.getReference()+1,atomic.getStamp(),atomic.getStamp()+1);
		File[] files = new File[0];
		try {
			File file = new File(localPath);
			files = file.listFiles();
			for (int i = 0; i < files.length; i++) {
				if (!uploadFiles(files[i], remotePath,fileName)) {
					return false;
				}
			}
		} catch (FtpException e) {
			throw e;
		}finally {
			atomic.compareAndSet(atomic.getReference(),atomic.getReference()-1,atomic.getStamp(),atomic.getStamp()+1);
			ftpsTimeOut.put(configFtp.getFtpId(),System.currentTimeMillis());
		}
		return files.length > 0;

	}

	/**
	 * 递归上传文件
	 *
	 * @param localeFile 本地文件/目录
	 * @param remotePath 上传目录
	 * @return boolean
	 * @throws FtpException e
	 */
	public boolean uploadFiles(File localeFile, String remotePath,String fileName) throws FtpException {
		atomic.compareAndSet(atomic.getReference(),atomic.getReference()+1,atomic.getStamp(),atomic.getStamp()+1);
		FileInputStream fis = null;
			try {
				if (localeFile.isDirectory()) {
					boolean flag = false;
					ftpClient.makeDirectory(remotePath);
					ftpClient.changeWorkingDirectory(remotePath);
					logger.info("[" + localeFile.getAbsolutePath() + "]目录");
					File[] files = localeFile.listFiles();
					for (File file : files) {
						if (uploadFiles(file, remotePath , localeFile.getName())) {
							flag = true;
						} else {
							return false;
						}
					}
					ftpClient.changeToParentDirectory();// return parent
					// directory
					return flag;
				} else if (!FILELOCK.equals(localeFile.getName())) {
					ftpClient.changeWorkingDirectory("/");
					this.createDirecroty(remotePath);
					ftpClient.changeWorkingDirectory(remotePath);
					fis = new FileInputStream(localeFile);
					ftpClient.storeFile(fileName, fis);
					logger.info("[" + localeFile.getAbsolutePath() + "]上传成功!");
					return true;
				}
				return true;
			} catch (IOException e) {
				logger.error("", e);
				throw new FtpException("FTP上传[" + localeFile.getAbsolutePath() + "]出错!", e);
			} finally {
				atomic.compareAndSet(atomic.getReference(),atomic.getReference()-1,atomic.getStamp(),atomic.getStamp()+1);
				ftpsTimeOut.put(configFtp.getFtpId(),System.currentTimeMillis());
				if (fis != null) {
					try {
						fis.close();
					} catch (IOException e) {
					}
				}
			}

	}

	/**
	 * 下载
	 *
	 * @param remotePath 下载目录
	 * @param localPath 本地目录
	 * @return boolean
	 * @throws FtpException e
	 */
	public boolean downLoadFile(String remotePath, String localPath) throws FtpException {
			try {
				if (ftpClient.changeWorkingDirectory(remotePath)) {// 转移到FTP服务器目录
					FTPFile[] files = ftpClient.listFiles();
					if (files.length > 0) {
						File localdir = new File(localPath);
						if (!localdir.exists()) {
							localdir.mkdir();
						}
					}
					for (FTPFile ff : files) {
						if (!downLoadFile(ff, localPath)) {
							return false;
						}
					}
					return files.length > 0;
				}
			} catch (IOException e) {
				logger.error("", e);
				throw new FtpException("FTP下载[" + localPath + "]出错!", e);
			}finally {
				atomic.compareAndSet(atomic.getReference(),atomic.getReference()-1,atomic.getStamp(),atomic.getStamp()+1);
				ftpsTimeOut.put(configFtp.getFtpId(),System.currentTimeMillis());
			}
			return false;
	}

	/**
	 * 递归下载文件
	 *
	 * @param ftpFile 下载文件/目录
	 * @param localPath 本地目录
	 * @return boolean
	 */
	public boolean downLoadFile(FTPFile ftpFile, String localPath) {
		// 当前处理文件本地路径
		String fileLocalPath = localPath + "/" + ftpFile.getName();
		if (ftpFile.isFile()) {// down file
			if (ftpFile.getName().indexOf("?") == -1) {
				OutputStream outputStream = null;
				atomic.compareAndSet(atomic.getReference(),atomic.getReference()+1,atomic.getStamp(),atomic.getStamp()+1);
				try {
					File localFile = new File(fileLocalPath);
					if (!localFile.getParentFile().exists()) {
						localFile.getParentFile().mkdirs();
					}
					outputStream = new FileOutputStream(localFile);
					ftpClient.retrieveFile(ftpFile.getName(), outputStream);
					outputStream.flush();
					outputStream.close();
					logger.info("[" + localFile.getAbsolutePath() + "]下载成功!");
					return true;
				} catch (Exception e) {
					logger.error("", e);
					throw new FtpException("FTP下载[" + fileLocalPath + "]出错!", e);
				} finally {
					atomic.compareAndSet(atomic.getReference(),atomic.getReference()-1,atomic.getStamp(),atomic.getStamp()+1);
					ftpsTimeOut.put(configFtp.getFtpId(),System.currentTimeMillis());
					try {
						if (outputStream != null) {
							outputStream.close();
						}
					} catch (IOException e) {
					}
				}
			}
		} else { // deal dirctory
			File file = new File(fileLocalPath);
			if (!file.exists()) {
				file.mkdirs();
			}
			try {
				// enter relative workdirectory
				if (ftpClient.changeWorkingDirectory(ftpFile.getName())) {
					logger.info("[" + file.getAbsolutePath() + "]目录");
					FTPFile[] files = null;
					files = ftpClient.listFiles();
					for (FTPFile file2 : files) {
						downLoadFile(file2, fileLocalPath);
					}
					ftpClient.changeToParentDirectory();// return parent
					// directory
					return true;
				}
			} catch (Exception e) {
				logger.error("", e);
				throw new FtpException("FTP下载[" + fileLocalPath + "]出错!", e);
			}
		}
		return false;
	}

	/** 获得目录下最大文件名 */
	public String getMaxFileName(String remotePath) {
		try {
			atomic.compareAndSet(atomic.getReference(),atomic.getReference()+1,atomic.getStamp(),atomic.getStamp()+1);

			ftpClient.changeWorkingDirectory(remotePath);
			FTPFile[] files = ftpClient.listFiles();
			Arrays.sort(files, new Comparator<FTPFile>() {
				@Override
				public int compare(FTPFile o1, FTPFile o2) {
					return o2.getName().compareTo(o1.getName());
				}
			});
			return files[0].getName();
		} catch (IOException e) {
			logger.error("", e);
			throw new FtpException("FTP访问目录[" + remotePath + "]出错!", e);
		}finally {
			atomic.compareAndSet(atomic.getReference(),atomic.getReference()-1,atomic.getStamp(),atomic.getStamp()+1);
			ftpsTimeOut.put(configFtp.getFtpId(),System.currentTimeMillis());
		}
	}

	public boolean createDirecroty(String remote) throws IOException {
		boolean success = true;
		String directory = remote + "/";
//        String directory = remote.substring(0, remote.lastIndexOf("/") + 1);
		// 如果远程目录不存在，则递归创建远程服务器目录
		if (!directory.equalsIgnoreCase("/") && !changeWorkingDirectory(new String(directory))) {
			int start = 0;
			int end = 0;
			if (directory.startsWith("/")) {
				start = 1;
			} else {
				start = 0;
			}
			end = directory.indexOf("/", start);
			String path = "";
			String paths = "";
			while (true) {

				String subDirectory = new String(remote.substring(start, end).getBytes("GBK"), "iso-8859-1");
				path = path + "/" + subDirectory;
				if (!existFile(path)) {
					if (makeDirectory(subDirectory)) {
						changeWorkingDirectory(subDirectory);
					} else {
						logger.debug("创建目录[" + subDirectory + "]失败");
						changeWorkingDirectory(subDirectory);
					}
				} else {
					changeWorkingDirectory(subDirectory);
				}

				paths = paths + "/" + subDirectory;
				start = end + 1;
				end = directory.indexOf("/", start);
				// 检查所有目录是否创建完毕
				if (end <= start) {
					break;
				}
			}
		}
		return success;
	}
	public boolean makeDirectory(String dir) {
		boolean flag = true;
		try {
			flag = ftpClient.makeDirectory(dir);
			if (flag) {
				logger.debug("创建文件夹" + dir + " 成功！");

			} else {
				logger.debug("创建文件夹" + dir + " 失败！");
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		return flag;
	}

	public boolean existFile(String path) throws IOException {
		boolean flag = false;
		FTPFile[] ftpFileArr = ftpClient.listFiles(path);
		if (ftpFileArr.length > 0) {
			flag = true;
		}
		return flag;
	}
	public boolean changeWorkingDirectory(String directory) {
		boolean flag = true;
		try {
			flag = ftpClient.changeWorkingDirectory(directory);
			if (flag) {
				logger.debug("进入文件夹" + directory + " 成功！");

			} else {
				logger.debug("进入文件夹" + directory + " 失败！");
			}
		} catch (IOException ioe) {
			ioe.printStackTrace();
		}
		return flag;
	}
}